package com.yeming.flink.practice.source

import java.util.Properties

import com.yeming.flink.mudlog.utilclass.MyKafkaReader
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, KafkaDeserializationSchema}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.codehaus.jackson.map.deser.std.StringDeserializer

object KafkaSourceKeyValue {

  def main(args: Array[String]): Unit = {
    //初始化环境变量
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //修改并行度
    streamEnv.setParallelism(1)
    //设置数据源,读取HDFS文件
    val props = new Properties()

    props.setProperty("bootstrap.servers","f2:9092,f3:9092,f4:9092")
    props.setProperty("group.id","flink02")
    props.setProperty("key.deserializer",classOf[StringDeserializer].getName)
    props.setProperty("value.deserializer",classOf[StringDeserializer].getName)
    props.setProperty("auto.offset.reset","latest")

    val stream: DataStream[(String, String)] = streamEnv.addSource(new FlinkKafkaConsumer[(String, String)]("flinktest01", new MyKafkaReader, props))
    //Transformation
    //Sink
    stream.print()

    streamEnv.execute("KafkaSourceKeyValue")
  }

}
